package pub.cleangao.memm.registry;

import lombok.extern.slf4j.Slf4j;
import pub.cleangao.memm.loadbalance.LoadBalancer;
import pub.cleangao.memm.transport.message.RpcRequest;
import pub.cleangao.memm.util.ThreadPoolFactory;

import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.util.*;
import java.util.concurrent.*;


@Slf4j
public abstract class AbstractRpcServiceRegistry implements RpcServiceRegistry {


    protected static final Map<String, InetSocketAddress> localFastCache = new ConcurrentHashMap<>();
    protected static final Map<String, Set<String>> localCache = new HashMap<>();
    protected static final Map<String, Object> serviceImpls = new HashMap<>();
    protected static final List<String> unhealthy = new ArrayList<>();
    protected static final Map<String, Integer> unhealthyCount = new HashMap<>();
    protected static final String ETCD_KEY_PREFIX = "RPC:";
    protected static String bindIp = RpcRuntime.getConfigProperties().getBindIp();
    protected static final int port = RpcRuntime.getConfigProperties().getPort();
    protected static String hostPort = bindIp + ":" + port;
    protected static final ExecutorService async = ThreadPoolFactory.createThreadPoolIfAbsent("rpc-client");
    protected static final ScheduledExecutorService healthSchedule = Executors.newSingleThreadScheduledExecutor();
    protected static final ScheduledExecutorService cacheLoadSchedule = Executors.newSingleThreadScheduledExecutor();
    protected static volatile boolean ifCacheReload = false;
    protected static Map<String, RpcServiceConfig> rpcServiceConfigMap = new HashMap<>();

    static {
        // 不使用"0.0.0.0"注册数据，但可使用这个IP启动端口监听
        if ("0.0.0.0".equals(bindIp)) {
            try {
                bindIp = InetAddress.getLocalHost().getHostAddress();
                hostPort = bindIp + ":" + port;
            } catch (UnknownHostException e) {
                e.printStackTrace();
            }
        }
    }

    { // non static
        healthSchedule.scheduleAtFixedRate(() -> {
            for (String hostPort : unhealthy) {
                Integer count = unhealthyCount.get(hostPort);
                if (count == null) {
                    count = 0;
                }
                if (count > 10) {
                    removeRemoteNode(hostPort);
                    unhealthyCount.remove(hostPort);
                    unhealthy.remove(hostPort);
                    continue;
                }
                boolean ok = RpcRuntime.testConnection(hostPort);
                if (ok) {
                    unhealthyCount.remove(hostPort);
                    unhealthy.remove(hostPort);
                } else {
                    unhealthyCount.put(hostPort, ++count);
                }
            }
        }, 15L, 15L, TimeUnit.SECONDS);
        cacheLoadSchedule.scheduleAtFixedRate(() -> {
            if (ifCacheReload) {
                log.info("Etcd storage updated, reloading cache...");
                reloadLocalCache();
                ifCacheReload = false;
            }
        }, 15L, 5L, TimeUnit.SECONDS);
    }


    /**
     * 发布服务
     */
    @Override
    public void publishService(RpcServiceConfig service) {
        // 1. 本地实现的服务接口
        serviceImpls.put(service.getRpcServiceName(), service.getServiceImpl());
        rpcServiceConfigMap.put(service.getRpcServiceName(), service);
        // 2. 发布到注册中心
        String key = ETCD_KEY_PREFIX + service.getRpcServiceName();
        boolean exist = false;
        List<String> list = loadValByPrefix(key);
        if (list != null && !list.isEmpty()) {
            if (list.contains(hostPort)) {
                exist = true;
            }
        }
        if (!exist) {
            storageService(service);
        }
    }


    public void storageService(RpcServiceConfig service) {
        String exactKey = ETCD_KEY_PREFIX + service.getRpcServiceName() + ":" + bindIp + "_" + port;
        String value = hostPort;
        putKeyValue(exactKey, value);
    }


    public void removeRemoteNode(String hostPort) {
        List<KeyVal> list = loadKeyValByPrefix(ETCD_KEY_PREFIX);
        if (list == null || list.isEmpty()) {
            return;
        }
        for (KeyVal keyVal : list) {
            if (keyVal.getValue().equals(hostPort)) {
                deleteByKey(keyVal.getKey());
            }
        }
    }

    protected abstract List<KeyVal> loadKeyValByPrefix(String prefix);

    protected abstract void putKeyValue(String key, String value);

    protected abstract void deleteByKey(String key);


    @Override
    public void reloadLocalCache() {
        async.execute(() -> {
            List<KeyVal> list = loadKeyValByPrefix(ETCD_KEY_PREFIX);
            if (list == null || list.isEmpty()) {
                return;
            }
            Map<String, Set<String>> newCache = new HashMap<>();
            boolean existThisNode = false;
            for (KeyVal keyValue : list) {
                String key = keyValue.getKey();
                String value = keyValue.getValue();
                key = key.substring(ETCD_KEY_PREFIX.length());
                String[] keyAt0 = key.split(":");
                Set<String> set = newCache.get(keyAt0[0]);
                if (set == null) {
                    set = new HashSet<>();
                }
                set.add(value);
                if (hostPort.equals(value)) {
                    existThisNode = true;
                }
                newCache.put(keyAt0[0], set);
            }
            localCache.clear();
            localCache.putAll(newCache);
            localFastCache.clear();
            if (!existThisNode) {
                republishServices();
            }
            log.info("Rpc local registry-cache reloaded.");

        });
    }

    private void republishServices() {
        for (Map.Entry<String, RpcServiceConfig> entry : rpcServiceConfigMap.entrySet()) {
            publishService(entry.getValue());
        }
    }


    /**
     * 查找可用服务
     */
    @Override
    public InetSocketAddress lookupService(RpcRequest rpcRequest) {
        // 1. 首先使用本地缓存
        // i_g_v
        String serviceName = rpcRequest.getRpcServiceName();
        InetSocketAddress fast = localFastCache.get(serviceName);
        if (fast != null) {
            return fast;
        }
        Set<String> addresses = localCache.get(serviceName);
        if (addresses != null && !addresses.isEmpty()) {
            String hostPort = LoadBalancer.selectOne(addresses.stream().toList(), rpcRequest, unhealthy);
            InetSocketAddress address = fromHostPort(hostPort);
            localFastCache.put(serviceName, address);
            return address;
        }

        // 2. 本地缓存无数据查询注册中心
        String key = ETCD_KEY_PREFIX + serviceName;
        List<String> list = loadValByPrefix(key);
        String hostPort = LoadBalancer.selectOne(list, rpcRequest, unhealthy);
        InetSocketAddress address = fromHostPort(hostPort);
        localFastCache.put(serviceName, address);
        reloadLocalCache();
        return address;

    }


    /**
     * 根据前缀从远程server加载数据
     *
     * @param prefix 前缀
     */
    public abstract List<String> loadValByPrefix(String prefix);


    /**
     * 本地接口调研
     *
     * @param rpcRequest RPC请求
     * @return 本地实现接口的服务实例
     */
    @Override
    public Object getServiceImpl(RpcRequest rpcRequest) {
        return serviceImpls.get(rpcRequest.getRpcServiceName());
    }


    @Override
    public void unregister() {
        List<KeyVal> list = loadKeyValByPrefix(ETCD_KEY_PREFIX);
        if (list == null || list.isEmpty()) {
            return;
        }
        for (KeyVal keyValue : list) {
            String key = keyValue.getKey();
            key = key.substring(ETCD_KEY_PREFIX.length());
            String[] keyAt1 = key.split(":");
            String key1 = bindIp + "_" + port;
            if (key1.equals(keyAt1[1])) {
                deleteByKey(keyValue.getKey());
            }
        }
    }

    /**
     * 标记非健康连接
     *
     * @param hostPort ip:port
     */
    @Override
    public void markUnHealth(String hostPort) {
        localFastCache.clear();
        if (!unhealthy.contains(hostPort)) {
            unhealthy.add(hostPort);
        }
    }

    /**
     * 全部的服务节点
     */
    public Set<String> allServeNode() {
        List<String> list = loadValByPrefix(ETCD_KEY_PREFIX);
        list.add(hostPort);
        return new HashSet<>(list);
    }


    private InetSocketAddress fromHostPort(String hostPort) {
        if (!hostPort.contains(":")) {
            return null;
        }
        String[] hostPortArray = hostPort.split(":");
        if (hostPortArray.length < 2) {
            return null;
        }
        if (hostPortArray[0] == null || hostPortArray[0].isBlank()) {
            return null;
        }
        if (hostPortArray[1] == null || hostPortArray[1].isBlank()) {
            return null;
        }
        return new InetSocketAddress(hostPortArray[0].trim(), Integer.parseInt(hostPortArray[1].trim()));
    }

}
